package com.glsc.ngateway.platform.utils;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.glsc.ngateway.common.api.flowable.dto.task.message.AbstractFlowMessageDto;
import com.glsc.ngateway.common.api.common.ex.GatewayException;
import com.glsc.ngateway.platform.service.MailService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @author zzp
 * @date 2021/5/27
 * 消息队列工具类
 */
@Component
public class MqUtil {
    private static final Logger logger = LoggerFactory.getLogger(MqUtil.class);

    public static String KEY_MSG_TYPE="_msgType";

    public static String KEY_MSG_GROUP="_msgGroup";

    public static String KEY_MSG_ID="_msgId";

    //消息的类型分组
    public static String MSG_TYPE_SYNC_NEW_PROD2OA="SYNC_NEW_PROD2OA";

    public static String MSG_TYPE_SYNC_UPDATE_PROD2OA="SYNC_UPDATE_PROD2OA";

    public static String MSG_TYPE_FEE_CARRY_REMIND_RERUN="FEE_CARRY_REMIND_RERUN";

    public static String MSG_TYPE_OPEN_DAY_RULE_RERUN="OPEN_DAY_RULE_RERUN";

    public static String MSG_TYPE_ACCOUNT_END_DATE ="ACCOUNT_END_DATE";

    private static MailService mailService;

    @Resource(name = "mailService")
    public void setMailService(MailService mailService) {
        MqUtil.mailService = mailService;
    }

    private static AmqpTemplate rabbitTemplate;

    @Resource(name = "rabbitTemplate")
    public void setRabbitTemplate(AmqpTemplate rabbitTemplate) {
        MqUtil.rabbitTemplate = rabbitTemplate;
    }

    /**
     * 生成消息队列公共头
     */
    public static Map<String,String> genMqHeader() {

        Map<String, String> header = new HashMap<>();
        header.put(KEY_MSG_TYPE, "0");
        header.put(KEY_MSG_GROUP, "0");
        header.put(KEY_MSG_ID, UUID.randomUUID().toString());
        return header;
    }

    /**
     * @param msgJson 消息json对象
     * @return 如果消息类型为空，则返回空字符串，不会返回null
     */
    public static String getMsgTypeFromJson(JSONObject msgJson) {

        if (msgJson == null || msgJson.get(MqUtil.KEY_MSG_TYPE) == null
                || StrUtil.isEmpty(msgJson.get(MqUtil.KEY_MSG_TYPE).toString())
        ) {

            String errMsg = MessageFormat.format("收到消息{0}时，解析消息类型{1}为空，异常！", (msgJson == null ? "" : msgJson.toString()), MqUtil.KEY_MSG_TYPE);

            logger.error(errMsg);
            return "";
        } else {
            return msgJson.get(MqUtil.KEY_MSG_TYPE).toString();
        }
    }

    /**
     * @param msgJson 消息json对象
     * @return 如果消息类型为空，则返回空字符串，不会返回null
     */
    public static String getString(JSONObject msgJson,String key,String defaultValue) {

        if (msgJson == null || msgJson.get(key) == null
                || StrUtil.isEmpty(msgJson.get(key).toString())
        ) {

            return defaultValue;
        } else {
            return msgJson.get(key).toString();
        }
    }

    /**
     * 发送消息
     */
    public static <T extends AbstractFlowMessageDto> void sendMessage(String exchange, String routingKey, T message) {
        Message msg = MessageBuilder
                .withBody(JSONObject.toJSONString(message).getBytes(StandardCharsets.UTF_8))
                .build();
        logger.info("WorkFlowSender开始发送消息：context :  " + msg);
        rabbitTemplate.send(exchange, routingKey, msg);
    }

    /**
     * @param messageObject 消息json对象
     * @param channel       通道
     * @param ackFlag       消息处理是否已经确认处理ackFlag=false，调用basicNack，消息进入死信队列
     * @return 如果消息类型为空，则返回空字符串，不会返回null
     */
    public static void processAckFlag(Message messageObject, Channel channel, boolean ackFlag, String exceptionTitle, Exception exception) {
        try {
            if (messageObject == null || channel == null) {
                String errMsg = "处理消息确认时，消息对象或者channel为空";
                logger.error(errMsg);
                throw GatewayException.error(errMsg);
            }
            String msg = new String(messageObject.getBody());


            //消息发送是否异常,basicNack后，进入死信队列
            if (!ackFlag) {
                String errorMsg = MessageFormat.format("接收到消息处理，发生异常{0}，消息内容：{1}，异常原因：{2}"
                        , exceptionTitle, msg, exception == null ? "" : exception.getMessage());

                mailService.sendMailToAdmin("产品系统消息处理异常", errorMsg);

                logger.error(errorMsg, exception);


                channel.basicNack(messageObject.getMessageProperties().getDeliveryTag(), false, false);
            } else {
                channel.basicAck(messageObject.getMessageProperties().getDeliveryTag(), false);
            }
        } catch (Exception e) {
            logger.error("设置消息队列确认状态异常,", e);
        }
    }


}
